/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.dpl.demo;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import org.apache.avro.Protocol;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.EncoderFactory;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.InputStreamEntity;
import org.apache.http.impl.client.DefaultHttpClient;

public class Demo extends Thread {

    public static void sendData(int batchSize, int sendTime, String url) {
        long startTime = System.currentTimeMillis();

        HttpClient httpClient = new DefaultHttpClient();
        //短连接，注意释放
        try {
            for (int i = 0; i < sendTime; i++) {
                byte[] seObject = getBinaryData(batchSize);

                //http post
                HttpPost httppost = new HttpPost(url);
                InputStreamEntity reqEntity = new InputStreamEntity(new ByteArrayInputStream(seObject), -1);

                //注意以下选项
                reqEntity.setContentType("binary/octet-stream");
                reqEntity.setChunked(true);

                httppost.setEntity(reqEntity);
                HttpResponse response = httpClient.execute(httppost);

                System.out.println(response.getStatusLine());
                httppost.releaseConnection();
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {

            httpClient.getConnectionManager().shutdown();
        }

        long endTime = System.currentTimeMillis();
        System.out.println("use " + (endTime - startTime) + " ms ");
    }

    private static byte[] getBinaryData(int batchSize) {
        try {
            Protocol protocol = Protocol.parse(new File("docs.json"));
            Schema docsSchema = protocol.getType("docs");

            DatumWriter<GenericRecord> docsWriter = new GenericDatumWriter<GenericRecord>(docsSchema);
            ByteArrayOutputStream docssos = new ByteArrayOutputStream();
            BinaryEncoder docsbe = new EncoderFactory().binaryEncoder(docssos, null);

            GenericRecord docsRecord = new GenericData.Record(docsSchema);
            docsRecord.put("doc_schema_name", "tableName");
            GenericArray docSet = new GenericData.Array<ByteBuffer>(batchSize, docsSchema.getField("doc_set").schema());

            protocol = Protocol.parse(new File("doc.json"));
            Schema grSingleDocSchema = protocol.getType("doc");
            DatumWriter<GenericRecord> grWriter = new GenericDatumWriter<GenericRecord>(grSingleDocSchema);
            ByteArrayOutputStream grsos = new ByteArrayOutputStream();
            BinaryEncoder grSingleDocEncoder = new EncoderFactory().binaryEncoder(grsos, null);

            Map<String, String> userDesc = new HashMap<String, String>();
            Map<String, String> docDesc = new HashMap<String, String>();

            /*
             * 写明此字段在表结构中的数据类型
             */
            docDesc.put("field1", "string_s");
            docDesc.put("field2", "string_s");
            docDesc.put("field3", "int_s");
            docDesc.put("field4", "double_s");
            docDesc.put("field5", "float_s");
            docDesc.put("field6", "long_s");


            //多条数据放入一个docSet中（实际是一个Array）
            for (int i = 0; i < batchSize; i++) {
                GenericRecord singleRecord = new GenericData.Record(grSingleDocSchema);
                HashMap<String, String> strMapSet = new HashMap<String, String>();
                HashMap<String, Boolean> boolMapSet = new HashMap<String, Boolean>();
                HashMap<String, Integer> intMapSet = new HashMap<String, Integer>();
                HashMap<String, Byte> byteMapSet = new HashMap<String, Byte>();
                HashMap<String, Double> doubleMapSet = new HashMap<String, Double>();
                HashMap<String, Float> floatMapSet = new HashMap<String, Float>();
                HashMap<String, Long> longMapSet = new HashMap<String, Long>();

                //按照表结构生成一条记录,每条记录的每个字段按照数据类型放入对应的map中
                //例如; field1是strin类型，那么就把字段名和值放入strMapSet中
                strMapSet.put("field1", "1");
                strMapSet.put("field2", "2");
                intMapSet.put("field3", 1);
                doubleMapSet.put("field4", 123d);
                floatMapSet.put("field5", 1234f);
                longMapSet.put("field6", 12345L);

                if (!strMapSet.isEmpty()) {
                    singleRecord.put("string_s", strMapSet);
                }
                if (!boolMapSet.isEmpty()) {
                    singleRecord.put("boolean_s", boolMapSet);
                }
                if (!intMapSet.isEmpty()) {
                    singleRecord.put("int_s", intMapSet);
                }
                if (!byteMapSet.isEmpty()) {
                    singleRecord.put("bytes_s", byteMapSet);
                }
                if (!doubleMapSet.isEmpty()) {
                    singleRecord.put("double_s", doubleMapSet);
                }
                if (!floatMapSet.isEmpty()) {
                    singleRecord.put("float_s", floatMapSet);
                }
                if (!longMapSet.isEmpty()) {
                    singleRecord.put("long_s", longMapSet);
                }
                grWriter.write(singleRecord, grSingleDocEncoder);
                grSingleDocEncoder.flush();
                docSet.add(ByteBuffer.wrap(grsos.toByteArray()));
                grsos.reset();
                System.out.println("one record is " + singleRecord);
            }

            docsRecord.put("doc_set", docSet);
            docsRecord.put("doc_desc", docDesc);
            docsRecord.put("user_desc", userDesc);

            docsWriter.write(docsRecord, docsbe);
            docsbe.flush();
            return docssos.toByteArray();
        } catch (Exception ex) {
            ex.printStackTrace();
            return null;
        }
    }

    public static void main(String[] args) {
        Demo.sendData(1, 1, "http://192.168.11.97:8010/dataload/");
    }
}